package com.intct.func;

import com.alibaba.fastjson.JSONObject;
import com.intct.util.HBaseUtil;
import com.intct.util.RowkeyUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.Map;

/**
 * @author gufg
 * @since 2025-08-20 11:34
 */
public class HBaseSink extends RichSinkFunction<JSONObject> {
    private Connection connection = null;

    private String namespace;

    /**
     * 分区数小于等于1时，代表不做rowkey散列
     */
    private int numParations;

    public HBaseSink(String namespace, int numParations) {
        this.namespace = namespace;
        this.numParations = numParations;
    }
    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取HBase的连接
        connection = HBaseUtil.getConnection();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeConnection(connection);
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {

        // {"sinkTable":"dim_area_info","op":"r","sinkRowKey":"code","after":{"code":321203,"name":"江苏省泰州市高港区"},"sinkFamily":"info"}
        String op = jsonObj.getString("op");

        String sinkTable = jsonObj.getString("sinkTable");

        String sinkFamily = jsonObj.getString("sinkFamily");

        String sinkRowKey = jsonObj.getString("sinkRowKey");

        JSONObject afterObj = jsonObj.getJSONObject("after");

        String[] rowkeyColumns = sinkRowKey.split(",");
        String rowkey = "";
        for (String rowkeyColumn : rowkeyColumns) {
            rowkey = rowkey + afterObj.getString(rowkeyColumn);
        }

        // 分区数小于等于1代表不做rowkey散列
        if (numParations > 1) {
            // 对rowkey进行散列，与表预分区对接
            rowkey = RowkeyUtil.getRowkey(rowkey, numParations);
        }

        // 拼接Hbase 的put对象
        Put put = new Put(Bytes.toBytes(rowkey));
        // 将JSON中的alter做为列限定和值
        for (Map.Entry<String, Object> entry : afterObj.entrySet()) {
            put.addColumn(Bytes.toBytes(sinkFamily), Bytes.toBytes(entry.getKey()), Bytes.toBytes(String.valueOf(entry.getValue())));
        }

        // 在数仓配置表删除记录
        if ("d".equals(op)) {
            HBaseUtil.dropTable(connection, namespace, sinkTable);
        } else {
            HBaseUtil.createTable(connection, namespace, sinkTable, sinkFamily);
            HBaseUtil.save(connection, namespace, sinkTable, put);
        }
    }
}
